/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.client.consumer.rebalance;

import java.util.ArrayList;
import java.util.List;
import org.apache.rocketmq.common.message.MessageQueue;

/**
 * Average Hashing queue algorithm
 */
public class AllocateMessageQueueAveragely extends AbstractAllocateMessageQueueStrategy {

    @Override
    public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll,
        List<String> cidAll) {
        // 创建结果集合
        List<MessageQueue> result = new ArrayList<>();
        // 检查
        if (!check(consumerGroup, currentCID, mqAll, cidAll)) {
            return result;
        }
        // 当前消费者在id列表中的索引
        int index = cidAll.indexOf(currentCID);
        // mq数量对消费者数量取模
        int mod = mqAll.size() % cidAll.size();
        //
        int averageSize = mqAll.size() <= cidAll.size()
                    ?
                    1 // 如果mq个数少于消费者个数，则每个消费者最多分配1个mq
                    :
                    (mod > 0 && index < mod  // 否则，如果模存在，并且索引小于模，则当前消费者在平均值的基础上+1，否则不用+1
                            ?
                            mqAll.size() / cidAll.size() + 1
                            :
                            mqAll.size() / cidAll.size()
                    );

        // 计算当前消费者分配mq索引的起始索引
        int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
        // 计算分配多少个mq
        int range = Math.min(averageSize, mqAll.size() - startIndex);

        // 将mq索引放到集合中
        for (int i = 0; i < range; i++) {
            result.add(mqAll.get((startIndex + i) % mqAll.size()));
        }
        // 返回结果
        return result;
    }

    @Override
    public String getName() {
        return "AVG";
    }
}
